package net.bwie.realtime.dws.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.dws.log.bean.AreaControl;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;

public class TrafficMapFunction extends RichMapFunction<String, AreaControl> {

    ValueState<String> car = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<String> descriptor = new ValueStateDescriptor<>("car", String.class);

        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.days(1))
                .updateTtlOnCreateAndWrite()
                .neverReturnExpired()
                .build();
        descriptor.enableTimeToLive(ttlConfig);

        car = getRuntimeContext().getState(descriptor);
    }

    @Override
    public AreaControl map(String value) throws Exception {
        //解析json
        JSONObject jsonObject = JSON.parseObject(value);
        //获取区域
        String areaId = jsonObject.getString("areaId");
        //获取车辆
        String car = jsonObject.getString("car");

        //车辆数
        Long carCount = 0L;
        if (null == this.car.value()){
            carCount = 1L;
            this.car.update(car);
        }

        return new AreaControl(
                areaId,
                carCount,
                null,
                null
        );
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
